package com.dtruth.kafka.consumer.pool;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

public class Consumer implements Runnable {

	private KafkaStream<byte[], byte[]> kafkaStream;
	private Integer threadNumber;

	public Consumer(KafkaStream<byte[], byte[]> kafkaStream,
			Integer threadNumber) {
		this.threadNumber = threadNumber;
		this.kafkaStream = kafkaStream;
	}

	public void run() {
		ConsumerIterator<byte[], byte[]> it = kafkaStream.iterator();
		System.out.println("Created iterator " + it.toString()
				+ " thread number " + threadNumber);
		while (true) {

			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				break;
			}
			while (it.hasNext()) {
				System.out.println("Thread " + threadNumber + ": "
						+ new String(it.next().message()));
			}
		}
		System.out.println("Shutting down Thread: " + threadNumber);
	}
}